<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1" />
<meta name="generator" content="pdoc 0.10.0" />
<title>netmiko.scp_handler API documentation</title>
<meta name="description" content="Netmiko SCP operations …" />
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/sanitize.min.css" integrity="sha256-PK9q560IAAa6WVRRh76LtCaI8pjTJ2z11v0miyNNjrs=" crossorigin>
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/typography.min.css" integrity="sha256-7l/o7C8jubJiy74VsKTidCy1yBkRtiUGbVkYBylBqUg=" crossorigin>
<link rel="stylesheet preload" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/styles/github.min.css" crossorigin>
<style>:root{--highlight-color:#fe9}.flex{display:flex !important}body{line-height:1.5em}#content{padding:20px}#sidebar{padding:30px;overflow:hidden}#sidebar > *:last-child{margin-bottom:2cm}.http-server-breadcrumbs{font-size:130%;margin:0 0 15px 0}#footer{font-size:.75em;padding:5px 30px;border-top:1px solid #ddd;text-align:right}#footer p{margin:0 0 0 1em;display:inline-block}#footer p:last-child{margin-right:30px}h1,h2,h3,h4,h5{font-weight:300}h1{font-size:2.5em;line-height:1.1em}h2{font-size:1.75em;margin:1em 0 .50em 0}h3{font-size:1.4em;margin:25px 0 10px 0}h4{margin:0;font-size:105%}h1:target,h2:target,h3:target,h4:target,h5:target,h6:target{background:var(--highlight-color);padding:.2em 0}a{color:#058;text-decoration:none;transition:color .3s ease-in-out}a:hover{color:#e82}.title code{font-weight:bold}h2[id^="header-"]{margin-top:2em}.ident{color:#900}pre code{background:#f8f8f8;font-size:.8em;line-height:1.4em}code{background:#f2f2f1;padding:1px 4px;overflow-wrap:break-word}h1 code{background:transparent}pre{background:#f8f8f8;border:0;border-top:1px solid #ccc;border-bottom:1px solid #ccc;margin:1em 0;padding:1ex}#http-server-module-list{display:flex;flex-flow:column}#http-server-module-list div{display:flex}#http-server-module-list dt{min-width:10%}#http-server-module-list p{margin-top:0}.toc ul,#index{list-style-type:none;margin:0;padding:0}#index code{background:transparent}#index h3{border-bottom:1px solid #ddd}#index ul{padding:0}#index h4{margin-top:.6em;font-weight:bold}@media (min-width:200ex){#index .two-column{column-count:2}}@media (min-width:300ex){#index .two-column{column-count:3}}dl{margin-bottom:2em}dl dl:last-child{margin-bottom:4em}dd{margin:0 0 1em 3em}#header-classes + dl > dd{margin-bottom:3em}dd dd{margin-left:2em}dd p{margin:10px 0}.name{background:#eee;font-weight:bold;font-size:.85em;padding:5px 10px;display:inline-block;min-width:40%}.name:hover{background:#e0e0e0}dt:target .name{background:var(--highlight-color)}.name > span:first-child{white-space:nowrap}.name.class > span:nth-child(2){margin-left:.4em}.inherited{color:#999;border-left:5px solid #eee;padding-left:1em}.inheritance em{font-style:normal;font-weight:bold}.desc h2{font-weight:400;font-size:1.25em}.desc h3{font-size:1em}.desc dt code{background:inherit}.source summary,.git-link-div{color:#666;text-align:right;font-weight:400;font-size:.8em;text-transform:uppercase}.source summary > *{white-space:nowrap;cursor:pointer}.git-link{color:inherit;margin-left:1em}.source pre{max-height:500px;overflow:auto;margin:0}.source pre code{font-size:12px;overflow:visible}.hlist{list-style:none}.hlist li{display:inline}.hlist li:after{content:',\2002'}.hlist li:last-child:after{content:none}.hlist .hlist{display:inline;padding-left:1em}img{max-width:100%}td{padding:0 .5em}.admonition{padding:.1em .5em;margin-bottom:1em}.admonition-title{font-weight:bold}.admonition.note,.admonition.info,.admonition.important{background:#aef}.admonition.todo,.admonition.versionadded,.admonition.tip,.admonition.hint{background:#dfd}.admonition.warning,.admonition.versionchanged,.admonition.deprecated{background:#fd4}.admonition.error,.admonition.danger,.admonition.caution{background:lightpink}</style>
<style media="screen and (min-width: 700px)">@media screen and (min-width:700px){#sidebar{width:30%;height:100vh;overflow:auto;position:sticky;top:0}#content{width:70%;max-width:100ch;padding:3em 4em;border-left:1px solid #ddd}pre code{font-size:1em}.item .name{font-size:1em}main{display:flex;flex-direction:row-reverse;justify-content:flex-end}.toc ul ul,#index ul{padding-left:1.5em}.toc > ul > li{margin-top:.5em}}</style>
<style media="print">@media print{#sidebar h1{page-break-before:always}.source{display:none}}@media print{*{background:transparent !important;color:#000 !important;box-shadow:none !important;text-shadow:none !important}a[href]:after{content:" (" attr(href) ")";font-size:90%}a[href][title]:after{content:none}abbr[title]:after{content:" (" attr(title) ")"}.ir a:after,a[href^="javascript:"]:after,a[href^="#"]:after{content:""}pre,blockquote{border:1px solid #999;page-break-inside:avoid}thead{display:table-header-group}tr,img{page-break-inside:avoid}img{max-width:100% !important}@page{margin:0.5cm}p,h2,h3{orphans:3;widows:3}h1,h2,h3,h4,h5,h6{page-break-after:avoid}}</style>
<script defer src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/highlight.min.js" integrity="sha256-Uv3H6lx7dJmRfRvH8TH6kJD1TSK1aFcwgx+mdg3epi8=" crossorigin></script>
<script>window.addEventListener('DOMContentLoaded', () => hljs.initHighlighting())</script>
</head>
<body>
<main>
<article id="content">
<header>
<h1 class="title">Module <code>netmiko.scp_handler</code></h1>
</header>
<section id="section-intro">
<p>Netmiko SCP operations.</p>
<p>Supports file get and file put operations.</p>
<p>SCP requires a separate SSH connection for a control channel.</p>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">&#34;&#34;&#34;
Netmiko SCP operations.

Supports file get and file put operations.

SCP requires a separate SSH connection for a control channel.
&#34;&#34;&#34;
from typing import Callable, Optional, Any, Type
from typing import TYPE_CHECKING
from types import TracebackType
import re
import os
import hashlib

import scp
import sys

if TYPE_CHECKING:
    from netmiko.base_connection import BaseConnection


class SCPConn(object):
    &#34;&#34;&#34;
    Establish a secure copy channel to the remote network device.

    Must close the SCP connection to get the file to write to the remote filesystem
    &#34;&#34;&#34;

    def __init__(
        self,
        ssh_conn: &#34;BaseConnection&#34;,
        socket_timeout: float = 10.0,
        progress: Optional[Callable[..., Any]] = None,
        progress4: Optional[Callable[..., Any]] = None,
    ) -&gt; None:
        self.ssh_ctl_chan = ssh_conn
        self.socket_timeout = socket_timeout
        self.progress = progress
        self.progress4 = progress4
        self.establish_scp_conn()

    def establish_scp_conn(self) -&gt; None:
        &#34;&#34;&#34;Establish the secure copy connection.&#34;&#34;&#34;
        ssh_connect_params = self.ssh_ctl_chan._connect_params_dict()
        self.scp_conn = self.ssh_ctl_chan._build_ssh_client()
        self.scp_conn.connect(**ssh_connect_params)
        self.scp_client = scp.SCPClient(
            self.scp_conn.get_transport(),
            socket_timeout=self.socket_timeout,
            progress=self.progress,
            progress4=self.progress4,
        )

    def scp_transfer_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Put file using SCP (for backwards compatibility).&#34;&#34;&#34;
        self.scp_client.put(source_file, dest_file)

    def scp_get_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Get file using SCP.&#34;&#34;&#34;
        self.scp_client.get(source_file, dest_file)

    def scp_put_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Put file using SCP.&#34;&#34;&#34;
        self.scp_client.put(source_file, dest_file)

    def close(self) -&gt; None:
        &#34;&#34;&#34;Close the SCP connection.&#34;&#34;&#34;
        self.scp_conn.close()


class BaseFileTransfer(object):
    &#34;&#34;&#34;Class to manage SCP file transfer and associated SSH control channel.&#34;&#34;&#34;

    def __init__(
        self,
        ssh_conn: &#34;BaseConnection&#34;,
        source_file: str,
        dest_file: str,
        file_system: Optional[str] = None,
        direction: str = &#34;put&#34;,
        socket_timeout: float = 10.0,
        progress: Optional[Callable[..., Any]] = None,
        progress4: Optional[Callable[..., Any]] = None,
        hash_supported: bool = True,
    ) -&gt; None:
        self.ssh_ctl_chan = ssh_conn
        self.source_file = source_file
        self.dest_file = dest_file
        self.direction = direction
        self.socket_timeout = socket_timeout
        self.progress = progress
        self.progress4 = progress4

        auto_flag = (
            &#34;cisco_ios&#34; in ssh_conn.device_type
            or &#34;cisco_xe&#34; in ssh_conn.device_type
            or &#34;cisco_xr&#34; in ssh_conn.device_type
        )
        if not file_system:
            if auto_flag:
                self.file_system = self.ssh_ctl_chan._autodetect_fs()
            else:
                raise ValueError(&#34;Destination file system not specified&#34;)
        else:
            self.file_system = file_system

        if direction == &#34;put&#34;:
            self.source_md5 = self.file_md5(source_file) if hash_supported else None
            self.file_size = os.stat(source_file).st_size
        elif direction == &#34;get&#34;:
            self.source_md5 = (
                self.remote_md5(remote_file=source_file) if hash_supported else None
            )
            self.file_size = self.remote_file_size(remote_file=source_file)
        else:
            raise ValueError(&#34;Invalid direction specified&#34;)

    def __enter__(self) -&gt; &#34;BaseFileTransfer&#34;:
        &#34;&#34;&#34;Context manager setup&#34;&#34;&#34;
        self.establish_scp_conn()
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -&gt; None:
        &#34;&#34;&#34;Context manager cleanup.&#34;&#34;&#34;
        self.close_scp_chan()

    def establish_scp_conn(self) -&gt; None:
        &#34;&#34;&#34;Establish SCP connection.&#34;&#34;&#34;
        self.scp_conn = SCPConn(
            self.ssh_ctl_chan,
            socket_timeout=self.socket_timeout,
            progress=self.progress,
            progress4=self.progress4,
        )

    def close_scp_chan(self) -&gt; None:
        &#34;&#34;&#34;Close the SCP connection to the remote network device.&#34;&#34;&#34;
        self.scp_conn.close()
        del self.scp_conn

    def remote_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; int:
        &#34;&#34;&#34;Return space available on remote device.&#34;&#34;&#34;
        remote_cmd = f&#34;dir {self.file_system}&#34;
        remote_output = self.ssh_ctl_chan._send_command_str(remote_cmd)
        match = re.search(search_pattern, remote_output)
        if match:
            if &#34;kbytes&#34; in match.group(0) or &#34;Kbytes&#34; in match.group(0):
                return int(match.group(1)) * 1000
            return int(match.group(1))
        else:
            msg = (
                f&#34;pattern: {search_pattern} not detected in output:\n\n{remote_output}&#34;
            )
            raise ValueError(msg)

    def _remote_space_available_unix(self, search_pattern: str = &#34;&#34;) -&gt; int:
        &#34;&#34;&#34;Return space available on *nix system (BSD/Linux).&#34;&#34;&#34;
        self.ssh_ctl_chan._enter_shell()
        remote_cmd = f&#34;/bin/df -k {self.file_system}&#34;
        remote_output = self.ssh_ctl_chan._send_command_str(
            remote_cmd, expect_string=r&#34;[\$#]&#34;
        )

        # Try to ensure parsing is correct:
        # Filesystem   1K-blocks  Used   Avail Capacity  Mounted on
        # /dev/bo0s3f    1264808 16376 1147248     1%    /cf/var
        remote_output = remote_output.strip()
        output_lines = remote_output.splitlines()

        # First line is the header; second is the actual file system info
        header_line = output_lines[0]
        filesystem_line = output_lines[1]

        if &#34;Filesystem&#34; not in header_line or &#34;Avail&#34; not in header_line.split()[3]:
            # Filesystem  1K-blocks  Used   Avail Capacity  Mounted on
            msg = &#34;Parsing error, unexpected output from {}:\n{}&#34;.format(
                remote_cmd, remote_output
            )
            raise ValueError(msg)

        space_available = filesystem_line.split()[3]
        if not re.search(r&#34;^\d+$&#34;, space_available):
            msg = &#34;Parsing error, unexpected output from {}:\n{}&#34;.format(
                remote_cmd, remote_output
            )
            raise ValueError(msg)

        self.ssh_ctl_chan._return_cli()
        return int(space_available) * 1024

    def local_space_available(self) -&gt; int:
        &#34;&#34;&#34;Return space available on local filesystem.&#34;&#34;&#34;
        if sys.platform == &#34;win32&#34;:
            import ctypes

            free_bytes = ctypes.c_ulonglong(0)
            ctypes.windll.kernel32.GetDiskFreeSpaceExW(
                ctypes.c_wchar_p(&#34;.&#34;), None, None, ctypes.pointer(free_bytes)
            )
            return free_bytes.value
        else:
            destination_stats = os.statvfs(&#34;.&#34;)
            return destination_stats.f_bsize * destination_stats.f_bavail

    def verify_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; bool:
        &#34;&#34;&#34;Verify sufficient space is available on destination file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            space_avail = self.remote_space_available(search_pattern=search_pattern)
        elif self.direction == &#34;get&#34;:
            space_avail = self.local_space_available()
        if space_avail &gt; self.file_size:
            return True
        return False

    def check_file_exists(self, remote_cmd: str = &#34;&#34;) -&gt; bool:
        &#34;&#34;&#34;Check if the dest_file already exists on the file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            if not remote_cmd:
                remote_cmd = f&#34;dir {self.file_system}/{self.dest_file}&#34;
            remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
            search_string = r&#34;Directory of .*{0}&#34;.format(self.dest_file)
            if (
                &#34;Error opening&#34; in remote_out
                or &#34;No such file or directory&#34; in remote_out
                or &#34;Path does not exist&#34; in remote_out
            ):
                return False
            elif re.search(search_string, remote_out, flags=re.DOTALL):
                return True
            else:
                raise ValueError(&#34;Unexpected output from check_file_exists&#34;)
        elif self.direction == &#34;get&#34;:
            return os.path.exists(self.dest_file)
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def _check_file_exists_unix(self, remote_cmd: str = &#34;&#34;) -&gt; bool:
        &#34;&#34;&#34;Check if the dest_file already exists on the file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            self.ssh_ctl_chan._enter_shell()
            remote_cmd = f&#34;/bin/ls {self.file_system}&#34;
            remote_out = self.ssh_ctl_chan._send_command_str(
                remote_cmd, expect_string=r&#34;[\$#]&#34;
            )
            self.ssh_ctl_chan._return_cli()
            return self.dest_file in remote_out
        elif self.direction == &#34;get&#34;:
            return os.path.exists(self.dest_file)
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def remote_file_size(
        self, remote_cmd: str = &#34;&#34;, remote_file: Optional[str] = None
    ) -&gt; int:
        &#34;&#34;&#34;Get the file size of the remote file.&#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        if not remote_cmd:
            remote_cmd = f&#34;dir {self.file_system}/{remote_file}&#34;
        remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
        # Strip out &#34;Directory of flash:/filename line
        remote_out_lines = re.split(r&#34;Directory of .*&#34;, remote_out)
        remote_out = &#34;&#34;.join(remote_out_lines)
        # Match line containing file name
        assert isinstance(remote_file, str)
        escape_file_name = re.escape(remote_file)
        pattern = r&#34;.*({}).*&#34;.format(escape_file_name)
        match = re.search(pattern, remote_out)
        if match:
            line = match.group(0)
            # Format will be 26  -rw-   6738  Jul 30 2016 19:49:50 -07:00  filename
            file_size = line.split()[2]
        else:
            raise IOError(&#34;Unable to parse &#39;dir&#39; output in remote_file_size method&#34;)

        if &#34;Error opening&#34; in remote_out or &#34;No such file or directory&#34; in remote_out:
            raise IOError(&#34;Unable to find file on remote system&#34;)
        else:
            return int(file_size)

    def _remote_file_size_unix(
        self, remote_cmd: str = &#34;&#34;, remote_file: Optional[str] = None
    ) -&gt; int:
        &#34;&#34;&#34;Get the file size of the remote file.&#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        remote_file = f&#34;{self.file_system}/{remote_file}&#34;
        if not remote_cmd:
            remote_cmd = f&#34;/bin/ls -l {remote_file}&#34;

        self.ssh_ctl_chan._enter_shell()
        remote_out = self.ssh_ctl_chan._send_command_str(
            remote_cmd, expect_string=r&#34;[\$#]&#34;
        )
        self.ssh_ctl_chan._return_cli()

        if &#34;No such file or directory&#34; in remote_out:
            raise IOError(&#34;Unable to find file on remote system&#34;)

        escape_file_name = re.escape(remote_file)
        pattern = r&#34;^.* ({}).*$&#34;.format(escape_file_name)
        match = re.search(pattern, remote_out, flags=re.M)
        if match:
            # Format: -rw-r--r--  1 pyclass  wheel  12 Nov  5 19:07 /var/tmp/test3.txt
            line = match.group(0)
            file_size = line.split()[4]
            return int(file_size)

        raise ValueError(
            &#34;Search pattern not found for remote file size during SCP transfer.&#34;
        )

    def file_md5(self, file_name: str, add_newline: bool = False) -&gt; str:
        &#34;&#34;&#34;Compute MD5 hash of file.

        add_newline is needed to support Cisco IOS MD5 calculation which expects the newline in
        the string

        Args:
          file_name: name of file to get md5 digest of
          add_newline: add newline to end of file contents or not

        &#34;&#34;&#34;
        file_hash = hashlib.md5()
        with open(file_name, &#34;rb&#34;) as f:
            while True:
                file_contents = f.read(512)
                if not file_contents:
                    if add_newline:
                        file_contents + b&#34;\n&#34;
                    break
                file_hash.update(file_contents)
        return file_hash.hexdigest()

    @staticmethod
    def process_md5(md5_output: str, pattern: str = r&#34;=\s+(\S+)&#34;) -&gt; str:
        &#34;&#34;&#34;
        Process the string to retrieve the MD5 hash

        Output from Cisco IOS (ASA is similar)
        .MD5 of flash:file_name Done!
        verify /md5 (flash:file_name) = 410db2a7015eaa42b1fe71f1bf3d59a2
        &#34;&#34;&#34;
        match = re.search(pattern, md5_output)
        if match:
            return match.group(1)
        else:
            raise ValueError(f&#34;Invalid output from MD5 command: {md5_output}&#34;)

    def compare_md5(self) -&gt; bool:
        &#34;&#34;&#34;Compare md5 of file on network device to md5 of local file.&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            remote_md5 = self.remote_md5()
            return self.source_md5 == remote_md5
        elif self.direction == &#34;get&#34;:
            local_md5 = self.file_md5(self.dest_file)
            return self.source_md5 == local_md5
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def remote_md5(
        self, base_cmd: str = &#34;verify /md5&#34;, remote_file: Optional[str] = None
    ) -&gt; str:
        &#34;&#34;&#34;Calculate remote MD5 and returns the hash.

        This command can be CPU intensive on the remote device.
        &#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        remote_md5_cmd = f&#34;{base_cmd} {self.file_system}/{remote_file}&#34;
        dest_md5 = self.ssh_ctl_chan._send_command_str(remote_md5_cmd, read_timeout=300)
        dest_md5 = self.process_md5(dest_md5)
        return dest_md5

    def transfer_file(self) -&gt; None:
        &#34;&#34;&#34;SCP transfer file.&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            self.put_file()
        elif self.direction == &#34;get&#34;:
            self.get_file()
        else:
            raise ValueError(&#34;Unexpected value for self.direction in transfer_file&#34;)

    def get_file(self) -&gt; None:
        &#34;&#34;&#34;SCP copy the file from the remote device to local system.&#34;&#34;&#34;
        source_file = f&#34;{self.file_system}/{self.source_file}&#34;
        self.scp_conn.scp_get_file(source_file, self.dest_file)
        self.scp_conn.close()

    def put_file(self) -&gt; None:
        &#34;&#34;&#34;SCP copy the file from the local system to the remote device.&#34;&#34;&#34;
        destination = f&#34;{self.file_system}/{self.dest_file}&#34;
        self.scp_conn.scp_transfer_file(self.source_file, destination)
        # Must close the SCP connection to get the file written (flush)
        self.scp_conn.close()

    def verify_file(self) -&gt; bool:
        &#34;&#34;&#34;Verify the file has been transferred correctly.&#34;&#34;&#34;
        return self.compare_md5()

    def enable_scp(self, cmd: str = &#34;ip scp server enable&#34;) -&gt; None:
        &#34;&#34;&#34;
        Enable SCP on remote device.

        Defaults to Cisco IOS command
        &#34;&#34;&#34;
        self.ssh_ctl_chan.send_config_set(cmd)

    def disable_scp(self, cmd: str = &#34;no ip scp server enable&#34;) -&gt; None:
        &#34;&#34;&#34;
        Disable SCP on remote device.

        Defaults to Cisco IOS command
        &#34;&#34;&#34;
        self.ssh_ctl_chan.send_config_set(cmd)</code></pre>
</details>
</section>
<section>
</section>
<section>
</section>
<section>
</section>
<section>
<h2 class="section-title" id="header-classes">Classes</h2>
<dl>
<dt id="netmiko.scp_handler.BaseFileTransfer"><code class="flex name class">
<span>class <span class="ident">BaseFileTransfer</span></span>
<span>(</span><span>ssh_conn: BaseConnection, source_file: str, dest_file: str, file_system: Optional[str] = None, direction: str = 'put', socket_timeout: float = 10.0, progress: Optional[Callable[..., Any]] = None, progress4: Optional[Callable[..., Any]] = None, hash_supported: bool = True)</span>
</code></dt>
<dd>
<div class="desc"><p>Class to manage SCP file transfer and associated SSH control channel.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class BaseFileTransfer(object):
    &#34;&#34;&#34;Class to manage SCP file transfer and associated SSH control channel.&#34;&#34;&#34;

    def __init__(
        self,
        ssh_conn: &#34;BaseConnection&#34;,
        source_file: str,
        dest_file: str,
        file_system: Optional[str] = None,
        direction: str = &#34;put&#34;,
        socket_timeout: float = 10.0,
        progress: Optional[Callable[..., Any]] = None,
        progress4: Optional[Callable[..., Any]] = None,
        hash_supported: bool = True,
    ) -&gt; None:
        self.ssh_ctl_chan = ssh_conn
        self.source_file = source_file
        self.dest_file = dest_file
        self.direction = direction
        self.socket_timeout = socket_timeout
        self.progress = progress
        self.progress4 = progress4

        auto_flag = (
            &#34;cisco_ios&#34; in ssh_conn.device_type
            or &#34;cisco_xe&#34; in ssh_conn.device_type
            or &#34;cisco_xr&#34; in ssh_conn.device_type
        )
        if not file_system:
            if auto_flag:
                self.file_system = self.ssh_ctl_chan._autodetect_fs()
            else:
                raise ValueError(&#34;Destination file system not specified&#34;)
        else:
            self.file_system = file_system

        if direction == &#34;put&#34;:
            self.source_md5 = self.file_md5(source_file) if hash_supported else None
            self.file_size = os.stat(source_file).st_size
        elif direction == &#34;get&#34;:
            self.source_md5 = (
                self.remote_md5(remote_file=source_file) if hash_supported else None
            )
            self.file_size = self.remote_file_size(remote_file=source_file)
        else:
            raise ValueError(&#34;Invalid direction specified&#34;)

    def __enter__(self) -&gt; &#34;BaseFileTransfer&#34;:
        &#34;&#34;&#34;Context manager setup&#34;&#34;&#34;
        self.establish_scp_conn()
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -&gt; None:
        &#34;&#34;&#34;Context manager cleanup.&#34;&#34;&#34;
        self.close_scp_chan()

    def establish_scp_conn(self) -&gt; None:
        &#34;&#34;&#34;Establish SCP connection.&#34;&#34;&#34;
        self.scp_conn = SCPConn(
            self.ssh_ctl_chan,
            socket_timeout=self.socket_timeout,
            progress=self.progress,
            progress4=self.progress4,
        )

    def close_scp_chan(self) -&gt; None:
        &#34;&#34;&#34;Close the SCP connection to the remote network device.&#34;&#34;&#34;
        self.scp_conn.close()
        del self.scp_conn

    def remote_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; int:
        &#34;&#34;&#34;Return space available on remote device.&#34;&#34;&#34;
        remote_cmd = f&#34;dir {self.file_system}&#34;
        remote_output = self.ssh_ctl_chan._send_command_str(remote_cmd)
        match = re.search(search_pattern, remote_output)
        if match:
            if &#34;kbytes&#34; in match.group(0) or &#34;Kbytes&#34; in match.group(0):
                return int(match.group(1)) * 1000
            return int(match.group(1))
        else:
            msg = (
                f&#34;pattern: {search_pattern} not detected in output:\n\n{remote_output}&#34;
            )
            raise ValueError(msg)

    def _remote_space_available_unix(self, search_pattern: str = &#34;&#34;) -&gt; int:
        &#34;&#34;&#34;Return space available on *nix system (BSD/Linux).&#34;&#34;&#34;
        self.ssh_ctl_chan._enter_shell()
        remote_cmd = f&#34;/bin/df -k {self.file_system}&#34;
        remote_output = self.ssh_ctl_chan._send_command_str(
            remote_cmd, expect_string=r&#34;[\$#]&#34;
        )

        # Try to ensure parsing is correct:
        # Filesystem   1K-blocks  Used   Avail Capacity  Mounted on
        # /dev/bo0s3f    1264808 16376 1147248     1%    /cf/var
        remote_output = remote_output.strip()
        output_lines = remote_output.splitlines()

        # First line is the header; second is the actual file system info
        header_line = output_lines[0]
        filesystem_line = output_lines[1]

        if &#34;Filesystem&#34; not in header_line or &#34;Avail&#34; not in header_line.split()[3]:
            # Filesystem  1K-blocks  Used   Avail Capacity  Mounted on
            msg = &#34;Parsing error, unexpected output from {}:\n{}&#34;.format(
                remote_cmd, remote_output
            )
            raise ValueError(msg)

        space_available = filesystem_line.split()[3]
        if not re.search(r&#34;^\d+$&#34;, space_available):
            msg = &#34;Parsing error, unexpected output from {}:\n{}&#34;.format(
                remote_cmd, remote_output
            )
            raise ValueError(msg)

        self.ssh_ctl_chan._return_cli()
        return int(space_available) * 1024

    def local_space_available(self) -&gt; int:
        &#34;&#34;&#34;Return space available on local filesystem.&#34;&#34;&#34;
        if sys.platform == &#34;win32&#34;:
            import ctypes

            free_bytes = ctypes.c_ulonglong(0)
            ctypes.windll.kernel32.GetDiskFreeSpaceExW(
                ctypes.c_wchar_p(&#34;.&#34;), None, None, ctypes.pointer(free_bytes)
            )
            return free_bytes.value
        else:
            destination_stats = os.statvfs(&#34;.&#34;)
            return destination_stats.f_bsize * destination_stats.f_bavail

    def verify_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; bool:
        &#34;&#34;&#34;Verify sufficient space is available on destination file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            space_avail = self.remote_space_available(search_pattern=search_pattern)
        elif self.direction == &#34;get&#34;:
            space_avail = self.local_space_available()
        if space_avail &gt; self.file_size:
            return True
        return False

    def check_file_exists(self, remote_cmd: str = &#34;&#34;) -&gt; bool:
        &#34;&#34;&#34;Check if the dest_file already exists on the file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            if not remote_cmd:
                remote_cmd = f&#34;dir {self.file_system}/{self.dest_file}&#34;
            remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
            search_string = r&#34;Directory of .*{0}&#34;.format(self.dest_file)
            if (
                &#34;Error opening&#34; in remote_out
                or &#34;No such file or directory&#34; in remote_out
                or &#34;Path does not exist&#34; in remote_out
            ):
                return False
            elif re.search(search_string, remote_out, flags=re.DOTALL):
                return True
            else:
                raise ValueError(&#34;Unexpected output from check_file_exists&#34;)
        elif self.direction == &#34;get&#34;:
            return os.path.exists(self.dest_file)
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def _check_file_exists_unix(self, remote_cmd: str = &#34;&#34;) -&gt; bool:
        &#34;&#34;&#34;Check if the dest_file already exists on the file system (return boolean).&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            self.ssh_ctl_chan._enter_shell()
            remote_cmd = f&#34;/bin/ls {self.file_system}&#34;
            remote_out = self.ssh_ctl_chan._send_command_str(
                remote_cmd, expect_string=r&#34;[\$#]&#34;
            )
            self.ssh_ctl_chan._return_cli()
            return self.dest_file in remote_out
        elif self.direction == &#34;get&#34;:
            return os.path.exists(self.dest_file)
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def remote_file_size(
        self, remote_cmd: str = &#34;&#34;, remote_file: Optional[str] = None
    ) -&gt; int:
        &#34;&#34;&#34;Get the file size of the remote file.&#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        if not remote_cmd:
            remote_cmd = f&#34;dir {self.file_system}/{remote_file}&#34;
        remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
        # Strip out &#34;Directory of flash:/filename line
        remote_out_lines = re.split(r&#34;Directory of .*&#34;, remote_out)
        remote_out = &#34;&#34;.join(remote_out_lines)
        # Match line containing file name
        assert isinstance(remote_file, str)
        escape_file_name = re.escape(remote_file)
        pattern = r&#34;.*({}).*&#34;.format(escape_file_name)
        match = re.search(pattern, remote_out)
        if match:
            line = match.group(0)
            # Format will be 26  -rw-   6738  Jul 30 2016 19:49:50 -07:00  filename
            file_size = line.split()[2]
        else:
            raise IOError(&#34;Unable to parse &#39;dir&#39; output in remote_file_size method&#34;)

        if &#34;Error opening&#34; in remote_out or &#34;No such file or directory&#34; in remote_out:
            raise IOError(&#34;Unable to find file on remote system&#34;)
        else:
            return int(file_size)

    def _remote_file_size_unix(
        self, remote_cmd: str = &#34;&#34;, remote_file: Optional[str] = None
    ) -&gt; int:
        &#34;&#34;&#34;Get the file size of the remote file.&#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        remote_file = f&#34;{self.file_system}/{remote_file}&#34;
        if not remote_cmd:
            remote_cmd = f&#34;/bin/ls -l {remote_file}&#34;

        self.ssh_ctl_chan._enter_shell()
        remote_out = self.ssh_ctl_chan._send_command_str(
            remote_cmd, expect_string=r&#34;[\$#]&#34;
        )
        self.ssh_ctl_chan._return_cli()

        if &#34;No such file or directory&#34; in remote_out:
            raise IOError(&#34;Unable to find file on remote system&#34;)

        escape_file_name = re.escape(remote_file)
        pattern = r&#34;^.* ({}).*$&#34;.format(escape_file_name)
        match = re.search(pattern, remote_out, flags=re.M)
        if match:
            # Format: -rw-r--r--  1 pyclass  wheel  12 Nov  5 19:07 /var/tmp/test3.txt
            line = match.group(0)
            file_size = line.split()[4]
            return int(file_size)

        raise ValueError(
            &#34;Search pattern not found for remote file size during SCP transfer.&#34;
        )

    def file_md5(self, file_name: str, add_newline: bool = False) -&gt; str:
        &#34;&#34;&#34;Compute MD5 hash of file.

        add_newline is needed to support Cisco IOS MD5 calculation which expects the newline in
        the string

        Args:
          file_name: name of file to get md5 digest of
          add_newline: add newline to end of file contents or not

        &#34;&#34;&#34;
        file_hash = hashlib.md5()
        with open(file_name, &#34;rb&#34;) as f:
            while True:
                file_contents = f.read(512)
                if not file_contents:
                    if add_newline:
                        file_contents + b&#34;\n&#34;
                    break
                file_hash.update(file_contents)
        return file_hash.hexdigest()

    @staticmethod
    def process_md5(md5_output: str, pattern: str = r&#34;=\s+(\S+)&#34;) -&gt; str:
        &#34;&#34;&#34;
        Process the string to retrieve the MD5 hash

        Output from Cisco IOS (ASA is similar)
        .MD5 of flash:file_name Done!
        verify /md5 (flash:file_name) = 410db2a7015eaa42b1fe71f1bf3d59a2
        &#34;&#34;&#34;
        match = re.search(pattern, md5_output)
        if match:
            return match.group(1)
        else:
            raise ValueError(f&#34;Invalid output from MD5 command: {md5_output}&#34;)

    def compare_md5(self) -&gt; bool:
        &#34;&#34;&#34;Compare md5 of file on network device to md5 of local file.&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            remote_md5 = self.remote_md5()
            return self.source_md5 == remote_md5
        elif self.direction == &#34;get&#34;:
            local_md5 = self.file_md5(self.dest_file)
            return self.source_md5 == local_md5
        else:
            raise ValueError(&#34;Unexpected value for self.direction&#34;)

    def remote_md5(
        self, base_cmd: str = &#34;verify /md5&#34;, remote_file: Optional[str] = None
    ) -&gt; str:
        &#34;&#34;&#34;Calculate remote MD5 and returns the hash.

        This command can be CPU intensive on the remote device.
        &#34;&#34;&#34;
        if remote_file is None:
            if self.direction == &#34;put&#34;:
                remote_file = self.dest_file
            elif self.direction == &#34;get&#34;:
                remote_file = self.source_file
        remote_md5_cmd = f&#34;{base_cmd} {self.file_system}/{remote_file}&#34;
        dest_md5 = self.ssh_ctl_chan._send_command_str(remote_md5_cmd, read_timeout=300)
        dest_md5 = self.process_md5(dest_md5)
        return dest_md5

    def transfer_file(self) -&gt; None:
        &#34;&#34;&#34;SCP transfer file.&#34;&#34;&#34;
        if self.direction == &#34;put&#34;:
            self.put_file()
        elif self.direction == &#34;get&#34;:
            self.get_file()
        else:
            raise ValueError(&#34;Unexpected value for self.direction in transfer_file&#34;)

    def get_file(self) -&gt; None:
        &#34;&#34;&#34;SCP copy the file from the remote device to local system.&#34;&#34;&#34;
        source_file = f&#34;{self.file_system}/{self.source_file}&#34;
        self.scp_conn.scp_get_file(source_file, self.dest_file)
        self.scp_conn.close()

    def put_file(self) -&gt; None:
        &#34;&#34;&#34;SCP copy the file from the local system to the remote device.&#34;&#34;&#34;
        destination = f&#34;{self.file_system}/{self.dest_file}&#34;
        self.scp_conn.scp_transfer_file(self.source_file, destination)
        # Must close the SCP connection to get the file written (flush)
        self.scp_conn.close()

    def verify_file(self) -&gt; bool:
        &#34;&#34;&#34;Verify the file has been transferred correctly.&#34;&#34;&#34;
        return self.compare_md5()

    def enable_scp(self, cmd: str = &#34;ip scp server enable&#34;) -&gt; None:
        &#34;&#34;&#34;
        Enable SCP on remote device.

        Defaults to Cisco IOS command
        &#34;&#34;&#34;
        self.ssh_ctl_chan.send_config_set(cmd)

    def disable_scp(self, cmd: str = &#34;no ip scp server enable&#34;) -&gt; None:
        &#34;&#34;&#34;
        Disable SCP on remote device.

        Defaults to Cisco IOS command
        &#34;&#34;&#34;
        self.ssh_ctl_chan.send_config_set(cmd)</code></pre>
</details>
<h3>Subclasses</h3>
<ul class="hlist">
<li><a title="netmiko.ciena.ciena_saos.CienaSaosFileTransfer" href="ciena/ciena_saos.html#netmiko.ciena.ciena_saos.CienaSaosFileTransfer">CienaSaosFileTransfer</a></li>
<li><a title="netmiko.cisco_base_connection.CiscoFileTransfer" href="cisco_base_connection.html#netmiko.cisco_base_connection.CiscoFileTransfer">CiscoFileTransfer</a></li>
<li><a title="netmiko.dell.dell_os10_ssh.DellOS10FileTransfer" href="dell/dell_os10_ssh.html#netmiko.dell.dell_os10_ssh.DellOS10FileTransfer">DellOS10FileTransfer</a></li>
<li><a title="netmiko.extreme.extreme_exos.ExtremeExosFileTransfer" href="extreme/extreme_exos.html#netmiko.extreme.extreme_exos.ExtremeExosFileTransfer">ExtremeExosFileTransfer</a></li>
<li><a title="netmiko.juniper.juniper.JuniperFileTransfer" href="juniper/juniper.html#netmiko.juniper.juniper.JuniperFileTransfer">JuniperFileTransfer</a></li>
<li><a title="netmiko.mikrotik.mikrotik_ssh.MikrotikRouterOsFileTransfer" href="mikrotik/mikrotik_ssh.html#netmiko.mikrotik.mikrotik_ssh.MikrotikRouterOsFileTransfer">MikrotikRouterOsFileTransfer</a></li>
<li><a title="netmiko.nokia.nokia_sros.NokiaSrosFileTransfer" href="nokia/nokia_sros.html#netmiko.nokia.nokia_sros.NokiaSrosFileTransfer">NokiaSrosFileTransfer</a></li>
</ul>
<h3>Static methods</h3>
<dl>
<dt id="netmiko.scp_handler.BaseFileTransfer.process_md5"><code class="name flex">
<span>def <span class="ident">process_md5</span></span>(<span>md5_output: str, pattern: str = '=\\s+(\\S+)') ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>Process the string to retrieve the MD5 hash</p>
<p>Output from Cisco IOS (ASA is similar)
.MD5 of flash:file_name Done!
verify /md5 (flash:file_name) = 410db2a7015eaa42b1fe71f1bf3d59a2</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@staticmethod
def process_md5(md5_output: str, pattern: str = r&#34;=\s+(\S+)&#34;) -&gt; str:
    &#34;&#34;&#34;
    Process the string to retrieve the MD5 hash

    Output from Cisco IOS (ASA is similar)
    .MD5 of flash:file_name Done!
    verify /md5 (flash:file_name) = 410db2a7015eaa42b1fe71f1bf3d59a2
    &#34;&#34;&#34;
    match = re.search(pattern, md5_output)
    if match:
        return match.group(1)
    else:
        raise ValueError(f&#34;Invalid output from MD5 command: {md5_output}&#34;)</code></pre>
</details>
</dd>
</dl>
<h3>Methods</h3>
<dl>
<dt id="netmiko.scp_handler.BaseFileTransfer.check_file_exists"><code class="name flex">
<span>def <span class="ident">check_file_exists</span></span>(<span>self, remote_cmd: str = '') ‑> bool</span>
</code></dt>
<dd>
<div class="desc"><p>Check if the dest_file already exists on the file system (return boolean).</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def check_file_exists(self, remote_cmd: str = &#34;&#34;) -&gt; bool:
    &#34;&#34;&#34;Check if the dest_file already exists on the file system (return boolean).&#34;&#34;&#34;
    if self.direction == &#34;put&#34;:
        if not remote_cmd:
            remote_cmd = f&#34;dir {self.file_system}/{self.dest_file}&#34;
        remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
        search_string = r&#34;Directory of .*{0}&#34;.format(self.dest_file)
        if (
            &#34;Error opening&#34; in remote_out
            or &#34;No such file or directory&#34; in remote_out
            or &#34;Path does not exist&#34; in remote_out
        ):
            return False
        elif re.search(search_string, remote_out, flags=re.DOTALL):
            return True
        else:
            raise ValueError(&#34;Unexpected output from check_file_exists&#34;)
    elif self.direction == &#34;get&#34;:
        return os.path.exists(self.dest_file)
    else:
        raise ValueError(&#34;Unexpected value for self.direction&#34;)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.close_scp_chan"><code class="name flex">
<span>def <span class="ident">close_scp_chan</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Close the SCP connection to the remote network device.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def close_scp_chan(self) -&gt; None:
    &#34;&#34;&#34;Close the SCP connection to the remote network device.&#34;&#34;&#34;
    self.scp_conn.close()
    del self.scp_conn</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.compare_md5"><code class="name flex">
<span>def <span class="ident">compare_md5</span></span>(<span>self) ‑> bool</span>
</code></dt>
<dd>
<div class="desc"><p>Compare md5 of file on network device to md5 of local file.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def compare_md5(self) -&gt; bool:
    &#34;&#34;&#34;Compare md5 of file on network device to md5 of local file.&#34;&#34;&#34;
    if self.direction == &#34;put&#34;:
        remote_md5 = self.remote_md5()
        return self.source_md5 == remote_md5
    elif self.direction == &#34;get&#34;:
        local_md5 = self.file_md5(self.dest_file)
        return self.source_md5 == local_md5
    else:
        raise ValueError(&#34;Unexpected value for self.direction&#34;)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.disable_scp"><code class="name flex">
<span>def <span class="ident">disable_scp</span></span>(<span>self, cmd: str = 'no ip scp server enable') ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Disable SCP on remote device.</p>
<p>Defaults to Cisco IOS command</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def disable_scp(self, cmd: str = &#34;no ip scp server enable&#34;) -&gt; None:
    &#34;&#34;&#34;
    Disable SCP on remote device.

    Defaults to Cisco IOS command
    &#34;&#34;&#34;
    self.ssh_ctl_chan.send_config_set(cmd)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.enable_scp"><code class="name flex">
<span>def <span class="ident">enable_scp</span></span>(<span>self, cmd: str = 'ip scp server enable') ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Enable SCP on remote device.</p>
<p>Defaults to Cisco IOS command</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def enable_scp(self, cmd: str = &#34;ip scp server enable&#34;) -&gt; None:
    &#34;&#34;&#34;
    Enable SCP on remote device.

    Defaults to Cisco IOS command
    &#34;&#34;&#34;
    self.ssh_ctl_chan.send_config_set(cmd)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.establish_scp_conn"><code class="name flex">
<span>def <span class="ident">establish_scp_conn</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Establish SCP connection.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def establish_scp_conn(self) -&gt; None:
    &#34;&#34;&#34;Establish SCP connection.&#34;&#34;&#34;
    self.scp_conn = SCPConn(
        self.ssh_ctl_chan,
        socket_timeout=self.socket_timeout,
        progress=self.progress,
        progress4=self.progress4,
    )</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.file_md5"><code class="name flex">
<span>def <span class="ident">file_md5</span></span>(<span>self, file_name: str, add_newline: bool = False) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>Compute MD5 hash of file.</p>
<p>add_newline is needed to support Cisco IOS MD5 calculation which expects the newline in
the string</p>
<h2 id="args">Args</h2>
<dl>
<dt><strong><code>file_name</code></strong></dt>
<dd>name of file to get md5 digest of</dd>
<dt><strong><code>add_newline</code></strong></dt>
<dd>add newline to end of file contents or not</dd>
</dl></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def file_md5(self, file_name: str, add_newline: bool = False) -&gt; str:
    &#34;&#34;&#34;Compute MD5 hash of file.

    add_newline is needed to support Cisco IOS MD5 calculation which expects the newline in
    the string

    Args:
      file_name: name of file to get md5 digest of
      add_newline: add newline to end of file contents or not

    &#34;&#34;&#34;
    file_hash = hashlib.md5()
    with open(file_name, &#34;rb&#34;) as f:
        while True:
            file_contents = f.read(512)
            if not file_contents:
                if add_newline:
                    file_contents + b&#34;\n&#34;
                break
            file_hash.update(file_contents)
    return file_hash.hexdigest()</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.get_file"><code class="name flex">
<span>def <span class="ident">get_file</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>SCP copy the file from the remote device to local system.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def get_file(self) -&gt; None:
    &#34;&#34;&#34;SCP copy the file from the remote device to local system.&#34;&#34;&#34;
    source_file = f&#34;{self.file_system}/{self.source_file}&#34;
    self.scp_conn.scp_get_file(source_file, self.dest_file)
    self.scp_conn.close()</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.local_space_available"><code class="name flex">
<span>def <span class="ident">local_space_available</span></span>(<span>self) ‑> int</span>
</code></dt>
<dd>
<div class="desc"><p>Return space available on local filesystem.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def local_space_available(self) -&gt; int:
    &#34;&#34;&#34;Return space available on local filesystem.&#34;&#34;&#34;
    if sys.platform == &#34;win32&#34;:
        import ctypes

        free_bytes = ctypes.c_ulonglong(0)
        ctypes.windll.kernel32.GetDiskFreeSpaceExW(
            ctypes.c_wchar_p(&#34;.&#34;), None, None, ctypes.pointer(free_bytes)
        )
        return free_bytes.value
    else:
        destination_stats = os.statvfs(&#34;.&#34;)
        return destination_stats.f_bsize * destination_stats.f_bavail</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.put_file"><code class="name flex">
<span>def <span class="ident">put_file</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>SCP copy the file from the local system to the remote device.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def put_file(self) -&gt; None:
    &#34;&#34;&#34;SCP copy the file from the local system to the remote device.&#34;&#34;&#34;
    destination = f&#34;{self.file_system}/{self.dest_file}&#34;
    self.scp_conn.scp_transfer_file(self.source_file, destination)
    # Must close the SCP connection to get the file written (flush)
    self.scp_conn.close()</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.remote_file_size"><code class="name flex">
<span>def <span class="ident">remote_file_size</span></span>(<span>self, remote_cmd: str = '', remote_file: Optional[str] = None) ‑> int</span>
</code></dt>
<dd>
<div class="desc"><p>Get the file size of the remote file.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def remote_file_size(
    self, remote_cmd: str = &#34;&#34;, remote_file: Optional[str] = None
) -&gt; int:
    &#34;&#34;&#34;Get the file size of the remote file.&#34;&#34;&#34;
    if remote_file is None:
        if self.direction == &#34;put&#34;:
            remote_file = self.dest_file
        elif self.direction == &#34;get&#34;:
            remote_file = self.source_file
    if not remote_cmd:
        remote_cmd = f&#34;dir {self.file_system}/{remote_file}&#34;
    remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
    # Strip out &#34;Directory of flash:/filename line
    remote_out_lines = re.split(r&#34;Directory of .*&#34;, remote_out)
    remote_out = &#34;&#34;.join(remote_out_lines)
    # Match line containing file name
    assert isinstance(remote_file, str)
    escape_file_name = re.escape(remote_file)
    pattern = r&#34;.*({}).*&#34;.format(escape_file_name)
    match = re.search(pattern, remote_out)
    if match:
        line = match.group(0)
        # Format will be 26  -rw-   6738  Jul 30 2016 19:49:50 -07:00  filename
        file_size = line.split()[2]
    else:
        raise IOError(&#34;Unable to parse &#39;dir&#39; output in remote_file_size method&#34;)

    if &#34;Error opening&#34; in remote_out or &#34;No such file or directory&#34; in remote_out:
        raise IOError(&#34;Unable to find file on remote system&#34;)
    else:
        return int(file_size)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.remote_md5"><code class="name flex">
<span>def <span class="ident">remote_md5</span></span>(<span>self, base_cmd: str = 'verify /md5', remote_file: Optional[str] = None) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>Calculate remote MD5 and returns the hash.</p>
<p>This command can be CPU intensive on the remote device.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def remote_md5(
    self, base_cmd: str = &#34;verify /md5&#34;, remote_file: Optional[str] = None
) -&gt; str:
    &#34;&#34;&#34;Calculate remote MD5 and returns the hash.

    This command can be CPU intensive on the remote device.
    &#34;&#34;&#34;
    if remote_file is None:
        if self.direction == &#34;put&#34;:
            remote_file = self.dest_file
        elif self.direction == &#34;get&#34;:
            remote_file = self.source_file
    remote_md5_cmd = f&#34;{base_cmd} {self.file_system}/{remote_file}&#34;
    dest_md5 = self.ssh_ctl_chan._send_command_str(remote_md5_cmd, read_timeout=300)
    dest_md5 = self.process_md5(dest_md5)
    return dest_md5</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.remote_space_available"><code class="name flex">
<span>def <span class="ident">remote_space_available</span></span>(<span>self, search_pattern: str = '(\\d+) \\w+ free') ‑> int</span>
</code></dt>
<dd>
<div class="desc"><p>Return space available on remote device.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def remote_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; int:
    &#34;&#34;&#34;Return space available on remote device.&#34;&#34;&#34;
    remote_cmd = f&#34;dir {self.file_system}&#34;
    remote_output = self.ssh_ctl_chan._send_command_str(remote_cmd)
    match = re.search(search_pattern, remote_output)
    if match:
        if &#34;kbytes&#34; in match.group(0) or &#34;Kbytes&#34; in match.group(0):
            return int(match.group(1)) * 1000
        return int(match.group(1))
    else:
        msg = (
            f&#34;pattern: {search_pattern} not detected in output:\n\n{remote_output}&#34;
        )
        raise ValueError(msg)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.transfer_file"><code class="name flex">
<span>def <span class="ident">transfer_file</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>SCP transfer file.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def transfer_file(self) -&gt; None:
    &#34;&#34;&#34;SCP transfer file.&#34;&#34;&#34;
    if self.direction == &#34;put&#34;:
        self.put_file()
    elif self.direction == &#34;get&#34;:
        self.get_file()
    else:
        raise ValueError(&#34;Unexpected value for self.direction in transfer_file&#34;)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.verify_file"><code class="name flex">
<span>def <span class="ident">verify_file</span></span>(<span>self) ‑> bool</span>
</code></dt>
<dd>
<div class="desc"><p>Verify the file has been transferred correctly.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def verify_file(self) -&gt; bool:
    &#34;&#34;&#34;Verify the file has been transferred correctly.&#34;&#34;&#34;
    return self.compare_md5()</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.BaseFileTransfer.verify_space_available"><code class="name flex">
<span>def <span class="ident">verify_space_available</span></span>(<span>self, search_pattern: str = '(\\d+) \\w+ free') ‑> bool</span>
</code></dt>
<dd>
<div class="desc"><p>Verify sufficient space is available on destination file system (return boolean).</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def verify_space_available(self, search_pattern: str = r&#34;(\d+) \w+ free&#34;) -&gt; bool:
    &#34;&#34;&#34;Verify sufficient space is available on destination file system (return boolean).&#34;&#34;&#34;
    if self.direction == &#34;put&#34;:
        space_avail = self.remote_space_available(search_pattern=search_pattern)
    elif self.direction == &#34;get&#34;:
        space_avail = self.local_space_available()
    if space_avail &gt; self.file_size:
        return True
    return False</code></pre>
</details>
</dd>
</dl>
</dd>
<dt id="netmiko.scp_handler.SCPConn"><code class="flex name class">
<span>class <span class="ident">SCPConn</span></span>
<span>(</span><span>ssh_conn: BaseConnection, socket_timeout: float = 10.0, progress: Optional[Callable[..., Any]] = None, progress4: Optional[Callable[..., Any]] = None)</span>
</code></dt>
<dd>
<div class="desc"><p>Establish a secure copy channel to the remote network device.</p>
<p>Must close the SCP connection to get the file to write to the remote filesystem</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class SCPConn(object):
    &#34;&#34;&#34;
    Establish a secure copy channel to the remote network device.

    Must close the SCP connection to get the file to write to the remote filesystem
    &#34;&#34;&#34;

    def __init__(
        self,
        ssh_conn: &#34;BaseConnection&#34;,
        socket_timeout: float = 10.0,
        progress: Optional[Callable[..., Any]] = None,
        progress4: Optional[Callable[..., Any]] = None,
    ) -&gt; None:
        self.ssh_ctl_chan = ssh_conn
        self.socket_timeout = socket_timeout
        self.progress = progress
        self.progress4 = progress4
        self.establish_scp_conn()

    def establish_scp_conn(self) -&gt; None:
        &#34;&#34;&#34;Establish the secure copy connection.&#34;&#34;&#34;
        ssh_connect_params = self.ssh_ctl_chan._connect_params_dict()
        self.scp_conn = self.ssh_ctl_chan._build_ssh_client()
        self.scp_conn.connect(**ssh_connect_params)
        self.scp_client = scp.SCPClient(
            self.scp_conn.get_transport(),
            socket_timeout=self.socket_timeout,
            progress=self.progress,
            progress4=self.progress4,
        )

    def scp_transfer_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Put file using SCP (for backwards compatibility).&#34;&#34;&#34;
        self.scp_client.put(source_file, dest_file)

    def scp_get_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Get file using SCP.&#34;&#34;&#34;
        self.scp_client.get(source_file, dest_file)

    def scp_put_file(self, source_file: str, dest_file: str) -&gt; None:
        &#34;&#34;&#34;Put file using SCP.&#34;&#34;&#34;
        self.scp_client.put(source_file, dest_file)

    def close(self) -&gt; None:
        &#34;&#34;&#34;Close the SCP connection.&#34;&#34;&#34;
        self.scp_conn.close()</code></pre>
</details>
<h3>Methods</h3>
<dl>
<dt id="netmiko.scp_handler.SCPConn.close"><code class="name flex">
<span>def <span class="ident">close</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Close the SCP connection.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def close(self) -&gt; None:
    &#34;&#34;&#34;Close the SCP connection.&#34;&#34;&#34;
    self.scp_conn.close()</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.SCPConn.establish_scp_conn"><code class="name flex">
<span>def <span class="ident">establish_scp_conn</span></span>(<span>self) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Establish the secure copy connection.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def establish_scp_conn(self) -&gt; None:
    &#34;&#34;&#34;Establish the secure copy connection.&#34;&#34;&#34;
    ssh_connect_params = self.ssh_ctl_chan._connect_params_dict()
    self.scp_conn = self.ssh_ctl_chan._build_ssh_client()
    self.scp_conn.connect(**ssh_connect_params)
    self.scp_client = scp.SCPClient(
        self.scp_conn.get_transport(),
        socket_timeout=self.socket_timeout,
        progress=self.progress,
        progress4=self.progress4,
    )</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.SCPConn.scp_get_file"><code class="name flex">
<span>def <span class="ident">scp_get_file</span></span>(<span>self, source_file: str, dest_file: str) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Get file using SCP.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def scp_get_file(self, source_file: str, dest_file: str) -&gt; None:
    &#34;&#34;&#34;Get file using SCP.&#34;&#34;&#34;
    self.scp_client.get(source_file, dest_file)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.SCPConn.scp_put_file"><code class="name flex">
<span>def <span class="ident">scp_put_file</span></span>(<span>self, source_file: str, dest_file: str) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Put file using SCP.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def scp_put_file(self, source_file: str, dest_file: str) -&gt; None:
    &#34;&#34;&#34;Put file using SCP.&#34;&#34;&#34;
    self.scp_client.put(source_file, dest_file)</code></pre>
</details>
</dd>
<dt id="netmiko.scp_handler.SCPConn.scp_transfer_file"><code class="name flex">
<span>def <span class="ident">scp_transfer_file</span></span>(<span>self, source_file: str, dest_file: str) ‑> None</span>
</code></dt>
<dd>
<div class="desc"><p>Put file using SCP (for backwards compatibility).</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def scp_transfer_file(self, source_file: str, dest_file: str) -&gt; None:
    &#34;&#34;&#34;Put file using SCP (for backwards compatibility).&#34;&#34;&#34;
    self.scp_client.put(source_file, dest_file)</code></pre>
</details>
</dd>
</dl>
</dd>
</dl>
</section>
</article>
<nav id="sidebar">
<h1>Index</h1>
<div class="toc">
<ul></ul>
</div>
<ul id="index">
<li><h3>Super-module</h3>
<ul>
<li><code><a title="netmiko" href="index.html">netmiko</a></code></li>
</ul>
</li>
<li><h3><a href="#header-classes">Classes</a></h3>
<ul>
<li>
<h4><code><a title="netmiko.scp_handler.BaseFileTransfer" href="#netmiko.scp_handler.BaseFileTransfer">BaseFileTransfer</a></code></h4>
<ul class="">
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.check_file_exists" href="#netmiko.scp_handler.BaseFileTransfer.check_file_exists">check_file_exists</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.close_scp_chan" href="#netmiko.scp_handler.BaseFileTransfer.close_scp_chan">close_scp_chan</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.compare_md5" href="#netmiko.scp_handler.BaseFileTransfer.compare_md5">compare_md5</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.disable_scp" href="#netmiko.scp_handler.BaseFileTransfer.disable_scp">disable_scp</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.enable_scp" href="#netmiko.scp_handler.BaseFileTransfer.enable_scp">enable_scp</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.establish_scp_conn" href="#netmiko.scp_handler.BaseFileTransfer.establish_scp_conn">establish_scp_conn</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.file_md5" href="#netmiko.scp_handler.BaseFileTransfer.file_md5">file_md5</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.get_file" href="#netmiko.scp_handler.BaseFileTransfer.get_file">get_file</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.local_space_available" href="#netmiko.scp_handler.BaseFileTransfer.local_space_available">local_space_available</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.process_md5" href="#netmiko.scp_handler.BaseFileTransfer.process_md5">process_md5</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.put_file" href="#netmiko.scp_handler.BaseFileTransfer.put_file">put_file</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.remote_file_size" href="#netmiko.scp_handler.BaseFileTransfer.remote_file_size">remote_file_size</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.remote_md5" href="#netmiko.scp_handler.BaseFileTransfer.remote_md5">remote_md5</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.remote_space_available" href="#netmiko.scp_handler.BaseFileTransfer.remote_space_available">remote_space_available</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.transfer_file" href="#netmiko.scp_handler.BaseFileTransfer.transfer_file">transfer_file</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.verify_file" href="#netmiko.scp_handler.BaseFileTransfer.verify_file">verify_file</a></code></li>
<li><code><a title="netmiko.scp_handler.BaseFileTransfer.verify_space_available" href="#netmiko.scp_handler.BaseFileTransfer.verify_space_available">verify_space_available</a></code></li>
</ul>
</li>
<li>
<h4><code><a title="netmiko.scp_handler.SCPConn" href="#netmiko.scp_handler.SCPConn">SCPConn</a></code></h4>
<ul class="">
<li><code><a title="netmiko.scp_handler.SCPConn.close" href="#netmiko.scp_handler.SCPConn.close">close</a></code></li>
<li><code><a title="netmiko.scp_handler.SCPConn.establish_scp_conn" href="#netmiko.scp_handler.SCPConn.establish_scp_conn">establish_scp_conn</a></code></li>
<li><code><a title="netmiko.scp_handler.SCPConn.scp_get_file" href="#netmiko.scp_handler.SCPConn.scp_get_file">scp_get_file</a></code></li>
<li><code><a title="netmiko.scp_handler.SCPConn.scp_put_file" href="#netmiko.scp_handler.SCPConn.scp_put_file">scp_put_file</a></code></li>
<li><code><a title="netmiko.scp_handler.SCPConn.scp_transfer_file" href="#netmiko.scp_handler.SCPConn.scp_transfer_file">scp_transfer_file</a></code></li>
</ul>
</li>
</ul>
</li>
</ul>
</nav>
</main>
<footer id="footer">
<p>Generated by <a href="https://pdoc3.github.io/pdoc" title="pdoc: Python API documentation generator"><cite>pdoc</cite> 0.10.0</a>.</p>
</footer>
</body>
</html>